package com.bjsdzk.collector.one2one;

import com.bjsdzk.backend.rpc.before.OneNETbefore;
import com.bjsdzk.collector.constant.Constant;
import com.bjsdzk.collector.entity.agreement.CallBackCmd;
import com.bjsdzk.collector.entity.agreement.SendCmd;
import com.bjsdzk.collector.serviceImp.DictionariesService;
import com.bjsdzk.collector.serviceImp.ExceColleContrService;
import com.bjsdzk.collector.serviceImp.ParamsSettingService;
import com.bjsdzk.collector.utils.Utils;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetServer;
import io.vertx.ext.mongo.BulkOperation;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.redis.RedisClient;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.stream.Collectors;

/**
 * @author pc
 */
public final class ZhuowenYanshiVerticleServer extends BasetVerticle {
    private static Logger logger = LoggerFactory.getLogger(ZhuowenYanshiVerticleServer.class);
    private OneNETbefore oneNETbefore;
    private boolean sendRpc=true;
    String imei = "860765046017666";
    Map<String, Long> keepLiveMap = new ConcurrentHashMap<>(16);

    public ZhuowenYanshiVerticleServer(RedisClient redis, MongoClient mongoClient, ExceColleContrService exceColleContrService, ParamsSettingService paramsSettingService, DictionariesService dictionariesService, OneNETbefore oneNETbefore) {
        this.redis = redis;
        this.mongoClient = mongoClient;
        this.exceColleContrService = exceColleContrService;
        this.paramsSettingService = paramsSettingService;
        this.dictionariesService = dictionariesService;
        this.oneNETbefore = oneNETbefore;
    }

    @Override
    public void start() {
        getInit();
        NetServer server = getNetServer();
        server.connectHandler(netSocket -> {
//            ConcurrentLinkedQueue<CallBackCmd> returnQueue = new ConcurrentLinkedQueue<>();
//            PriorityBlockingQueue<SendCmd> sendQueue = new PriorityBlockingQueue<>();
//            Map<String, Long> keepLiveMap = new ConcurrentHashMap<>(16);
            ReceiveServer receiveServer = new ReceiveServer();
            netSocket.handler(receiveServer);
        });

        long overtimeId = vertx.setPeriodic(5000, t -> {
            if (keepLiveMap.get("alive") != null
                    && (System.currentTimeMillis() - keepLiveMap.get("alive") > 1000*60*3)) {
                logger.info("port:{},time:{},offline",port,keepLiveMap.get("alive"));
                keepLiveMap.remove("alive");
                if (sendRpc){
                    oneNETbefore.changeState(imei ,0);
                    oneNETbefore.alarm(imei, new JsonObject().put("imei", imei).put("type", 424).toString());
                }
            }
        });

        serverListen(server,redis,"mod_"+port);

    }

    @Override
    public void stop() {

    }

    class ReceiveServer implements Handler<Buffer> {

        @Override
        public void handle(Buffer buffer) {
            byte[] bytes = buffer.getBytes();
            logger.info("port:{},dataHex:{}", port, Utils.bytes2String(bytes));
            if (sendRpc){
                oneNETbefore.changeState(imei ,1);
            }
            keepLiveMap.put("alive", System.currentTimeMillis());
            //0403DC000100210001000200000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000100000000000000000000006401F3014B000000000000000000000000000000000000000000EB0055010700A0490C005A001E0000000000000000000000000000000000000000000000000000002B004200FA00BE3E1C0046001E0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000BA45

            Integer deviceId = getValueBytype(bytes, 0);
            List<BulkOperation> bulkOperationList = new ArrayList<>();
            JsonArray lasteDate = new JsonArray();
            JsonObject baseJson = new JsonObject().put("deviceId", StringUtils.join(imei, "-", deviceId)).put("imei", imei).put("lineAddress", deviceId);

            saveMongoJson(bulkOperationList, lasteDate,baseJson, getValueBytype(bytes, 23, 2), 205);
            if (deviceId > 3) {
                int pos = 35*2+1;
                double electricCurrent = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, electricCurrent, 120);
                double voltage = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, voltage, 122);
                Integer electricLeakage = getValueBytype(bytes, pos += 2, 2);
                double activePower = getConversion(getValueBytype(bytes, pos += 2, 2),1000);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, activePower, 123);
                double reactivePower = getConversion(getValueBytype(bytes, pos += 2, 2),1000);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, reactivePower, 124);
                double electricQuantity = getConversion(getValueBytype(bytes, pos += 2, 4),10);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, electricQuantity, 4);
                double powerFactor = getConversion(getValueBytype(bytes, pos += 4, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, powerFactor, 126);
                double frequency = getConversion(getValueBytype(bytes, pos += 2, 2),10);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, frequency, 127);
                double temperature = getConversion(getValueBytype(bytes, pos += 2, 2),10);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, temperature, 121);

            } else {
                int pos = 95*2+1;
                double AelectricCurrent = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, AelectricCurrent, 12);
                double BelectricCurrent = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, BelectricCurrent, 13);
                double CelectricCurrent = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, CelectricCurrent, 14);
                double a_voltage = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, a_voltage, 9);
                double b_voltage = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, b_voltage, 10);
                double c_voltage = getConversion(getValueBytype(bytes, pos += 2, 2),100);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, c_voltage, 11);
                double a_activePower = getConversion(getValueBytype(bytes, pos += 2, 2));
                saveMongoJson(bulkOperationList, lasteDate,baseJson, a_activePower, 16);
                double b_activePower = getConversion(getValueBytype(bytes, pos += 2, 2));
                saveMongoJson(bulkOperationList, lasteDate,baseJson, b_activePower, 17);
                double c_activePower = getConversion(getValueBytype(bytes, pos += 2, 2));
                saveMongoJson(bulkOperationList, lasteDate,baseJson, c_activePower, 18);
                double a_reactivePower = getConversion(getValueBytype(bytes, pos += 2, 2));
                saveMongoJson(bulkOperationList, lasteDate,baseJson, a_reactivePower, 26);
                double b_reactivePower = getConversion(getValueBytype(bytes, pos += 2, 2));
                saveMongoJson(bulkOperationList, lasteDate,baseJson, b_reactivePower, 27);
                double c_reactivePower = getConversion(getValueBytype(bytes, pos += 2, 2));
                saveMongoJson(bulkOperationList, lasteDate,baseJson, c_reactivePower, 28);
                double a_temperature = getConversion(getValueBytype(bytes, pos += 2, 2),10);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, a_temperature, 130);
                double b_temperature = getConversion(getValueBytype(bytes, pos += 2, 2),10);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, b_temperature, 131);
                double c_temperature = getConversion(getValueBytype(bytes, pos += 2, 2),10);
                saveMongoJson(bulkOperationList, lasteDate,baseJson, c_temperature, 132);
            }
            if (!bulkOperationList.isEmpty()) {
                logger.info(lasteDate.toString());
                saveMongo(bulkOperationList, imei);
            }

            Integer alarm1 = getValueBytype(bytes, 49,2);
            sendAlarm(imei,155,deviceId,alarm1,1);
            sendAlarm(imei,219,deviceId,alarm1,2);
            sendAlarm(imei,221,deviceId,alarm1,4);
            sendAlarm(imei,213,deviceId,alarm1,8);
//            sendAlarm(imei,155,deviceId,alarm1,16);
            sendAlarm(imei,215,deviceId,alarm1,32);
            sendAlarm(imei,151,deviceId,alarm1,64);
            sendAlarm(imei,63,deviceId,alarm1,128);
            Integer alarm2 = getValueBytype(bytes, 51,2);
            sendAlarm(imei,12,deviceId,alarm2,1);
            sendAlarm(imei,91,deviceId,alarm2,2);
            sendAlarm(imei,92,deviceId,alarm2,4);
            sendAlarm(imei,226,deviceId,alarm2,8);
//            sendAlarm(imei,155,deviceId,alarm2,16);
            sendAlarm(imei,130,deviceId,alarm2,32);
            sendAlarm(imei,152,deviceId,alarm2,64);
//            sendAlarm(imei,63,deviceId,alarm2,128);
            Integer alarm3 = getValueBytype(bytes, 53,2);
            sendAlarm(imei,156,deviceId,alarm3,1);
            sendAlarm(imei,101,deviceId,alarm3,2);
            sendAlarm(imei,102,deviceId,alarm3,4);
            sendAlarm(imei,238,deviceId,alarm3,8);
//            sendAlarm(imei,155,deviceId,alarm4,16);
            sendAlarm(imei,131,deviceId,alarm3,32);
            sendAlarm(imei,153,deviceId,alarm3,64);
//            sendAlarm(imei,63,deviceId,alarm3,128);
            Integer alarm4 = getValueBytype(bytes, 55,2);
            sendAlarm(imei,14,deviceId,alarm4,1);
            sendAlarm(imei,111,deviceId,alarm4,2);
            sendAlarm(imei,112,deviceId,alarm4,4);
            sendAlarm(imei,250,deviceId,alarm4,8);
//            sendAlarm(imei,155,deviceId,alarm4,16);
            sendAlarm(imei,132,deviceId,alarm4,32);
            sendAlarm(imei,154,deviceId,alarm4,64);
//            sendAlarm(imei,63,deviceId,alarm3,128);
            Integer alarm5 = getValueBytype(bytes, 57,2);

            Integer police1 = getValueBytype(bytes, 59,2);
            sendPolice(imei,218,deviceId,police1,1);
            sendPolice(imei,220,deviceId,police1,2);
            sendPolice(imei,222,deviceId,police1,4);
            sendPolice(imei,214,deviceId,police1,8);
//            sendPolice(imei,4,deviceId,police1,16);
            sendPolice(imei,216,deviceId,police1,32);
//            sendPolice(imei,6,deviceId,police1,64);
            sendPolice(imei,212,deviceId,police1,128);
            Integer police2 = getValueBytype(bytes, 61,2);
            sendPolice(imei,229,deviceId,police2,1);
            sendPolice(imei,231,deviceId,police2,2);
            sendPolice(imei,233,deviceId,police2,4);
            sendPolice(imei,227,deviceId,police2,8);
//            sendPolice(imei,4,deviceId,police2,16);
            sendPolice(imei,235,deviceId,police2,32);
//            sendPolice(imei,6,deviceId,police2,64);
//            sendPolice(imei,2,deviceId,police2,128);
            Integer police3 = getValueBytype(bytes, 63,2);
            sendPolice(imei,241,deviceId,police3,1);
            sendPolice(imei,243,deviceId,police3,2);
            sendPolice(imei,245,deviceId,police3,4);
            sendPolice(imei,239,deviceId,police3,8);
//            sendPolice(imei,4,deviceId,police3,16);
            sendPolice(imei,423,deviceId,police3,32);
//            sendPolice(imei,6,deviceId,police3,64);
//            sendPolice(imei,2,deviceId,police3,128);
            Integer police4 = getValueBytype(bytes, 65,2);
            sendPolice(imei,252,deviceId,police4,1);
            sendPolice(imei,254,deviceId,police4,2);
            sendPolice(imei,256,deviceId,police4,4);
            sendPolice(imei,254,deviceId,police4,8);
//            sendPolice(imei,4,deviceId,police4,16);
            sendPolice(imei,258,deviceId,police4,32);
//            sendPolice(imei,6,deviceId,police4,64);
//            sendPolice(imei,2,deviceId,police4,128);

        }


        private double  getConversion(Integer value,Integer times){
            BigDecimal bd=new BigDecimal(value);
            BigDecimal divide = bd.divide(new BigDecimal(times), 2, BigDecimal.ROUND_HALF_UP);
            return divide.doubleValue();
        }

        private double  getConversion(Integer value){
            return getConversion(value,1000);
        }

        public void saveMongo( List<BulkOperation> bulkOperationList,String imei) {
            mongoClient.bulkWrite(Constant.MONGONAME_REAL, bulkOperationList, stringAsyncResult -> {
                if (stringAsyncResult.succeeded()) {
                    com.alibaba.fastjson.JSONArray data= bulkOperationList.stream().map((b) ->{
                        return b.getDocument().getString("_id");
                    }).collect(Collectors.toCollection(com.alibaba.fastjson.JSONArray::new));

                    try{
//                        logger.info(data.toString());
                        if (sendRpc){
                            oneNETbefore.data(imei, data.toString());
                        }
                    }catch (Exception e){
                        logger.error("onenet saveMongoAndSend rpc data invoke faile:{}",e.getMessage());
                    }
                } else {
                    logger.error("monggoDB onenet save faile： " + stringAsyncResult.cause().getMessage() +   "==>" );
                }

            });
        }
    }

    protected void sendPolice(String imei,Integer type,Integer deviceId,Integer policeFlag,Integer policeValue ) {
        if(isPolice(policeFlag, policeValue)){
            logger.info("port:{},police:"+new JsonObject().put("imei", imei).put("type", type).put("lineAddr", deviceId+"").toString(),port);
            if (sendRpc){
                oneNETbefore.police(imei, new JsonObject().put("imei", imei).put("type", type).put("lineAddr", deviceId+"").toString());
            }
        }
    }
    protected void sendAlarm(String imei,Integer type,Integer deviceId,Integer alarmFlag,Integer policeValue) {
        if(isPolice(alarmFlag, policeValue)) {
            logger.info("port:{},alarm:"+new JsonObject().put("imei", imei).put("type", type).put("lineAddr", deviceId+"").toString(),port);
            if (sendRpc){
                oneNETbefore.alarm(imei, new JsonObject().put("imei", imei).put("type", type).put("lineAddr", deviceId + "").toString());
            }
        }
    }
}

